package web

import (
	"context"
	"errors"
	"gitee.com/yysf_xin/dplog/pkg/config"
	"gitee.com/yysf_xin/dplog/pkg/extracter/fex"
	"gitee.com/yysf_xin/dplog/pkg/extracter/java"
	"gitee.com/yysf_xin/dplog/pkg/slog"
	"gitee.com/yysf_xin/dplog/pkg/web/pb"
	"gitee.com/yysf_xin/go-common/db"
	"gitee.com/yysf_xin/go-common/rest"
	"gitee.com/yysf_xin/go-common/tools"
	"github.com/go-chi/chi/v5"
	"github.com/gorilla/websocket"
	"google.golang.org/protobuf/encoding/protowire"
	"google.golang.org/protobuf/proto"
	"io"
	"io/fs"
	"log"
	"net/http"
	"os"
	"sort"
	"strings"
	"sync"
	"time"
)

const (
	evtReg int = iota
	evtAttach
)

const (
	// Time allowed to write a message to the peer.
	writeWait = 10 * time.Second

	// Time allowed to read the next pong message from the peer.
	pongWait = 60 * time.Second

	// Send pings to peer with this period. Must be less than pongWait.
	pingPeriod = (pongWait * 9) / 10

	readBufferSize  = 1024
	writeBufferSize = 1024 * 50
)

var (
	unknownService = errors.New("未知的服务")
)

type writePoolData struct{ buf []byte }

var (
	bufPool = sync.Pool{
		New: func() interface{} {
			return make([]byte, writeBufferSize)
		},
	}
)

var upgrader = websocket.Upgrader{
	ReadBufferSize:  readBufferSize,
	WriteBufferSize: writeBufferSize + 500,
	WriteBufferPool: &sync.Pool{},
	CheckOrigin: func(r *http.Request) bool {
		//data, err := httputil.DumpRequest(r, false)
		//if err != nil {
		//	log.Println(err)
		//} else {
		//	log.Println(string(data))
		//}
		return true
	},
}

type Cleaner func()

type client struct {
	s       *LogServer
	conn    *websocket.Conn
	lock    sync.Mutex
	req     pb.Request
	res     pb.Response
	cancels map[int32]Cleaner

	w     io.WriteCloser
	count int

	mx      sync.Mutex
	marshal proto.MarshalOptions
}

func (c *client) GetType() int {
	return evtReg
}
func (c *client) ping() error {
	c.mx.Lock()
	defer c.mx.Unlock()
	//_ = c.conn.SetWriteDeadline(time.Now().Add(writeWait))
	return c.conn.WriteMessage(websocket.PingMessage, nil)

}
func (c *client) Write(p []byte) (n int, err error) {
	c.mx.Lock()
	defer c.mx.Unlock()

	if c.w == nil {
		c.w, err = c.conn.NextWriter(websocket.BinaryMessage)
	}
	if c.w != nil {
		n, err = c.w.Write(p)
		c.count += n
		if c.count > writeBufferSize {
			c.Flush()
		}
	}
	return
}

func (c *client) Flush() {
	if c.w != nil {
		_ = c.w.Close()
		c.w = nil
		c.count = 0
	}
}

type Msg struct {
	Msg string `json:"msg"`
}

func (c *client) writeMsg() {
	t := time.NewTicker(pingPeriod)
	defer t.Stop()
	for {
		select {
		case <-t.C:
			err := c.ping()
			if err != nil {
				_ = c.conn.Close()
				return
			}
		}
	}
}
func (c *client) readMsg() {
	defer func() {
		for _, v := range c.cancels {
			v()
		}
		s := c.s
		c.s = nil
		s.evt <- c
		log.Println("conn closed", c.conn.RemoteAddr())
	}()
	log.Println("conn open", c.conn.RemoteAddr())
	c.conn.SetReadLimit(readBufferSize)

	f := func(string) error {
		_ = c.conn.SetReadDeadline(time.Now().Add(pongWait))
		return nil
	}
	_ = f("")

	c.conn.SetPongHandler(f)

	for {
		t, message, err := c.conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				log.Println("error: ", err)
			}
			break
		}

		if t != websocket.BinaryMessage {
			continue
		}

		err = proto.Unmarshal(message, &c.req)
		if err != nil {
			c.writeErr(err)
			continue
		}

		switch c.req.Type {
		case pb.MsgType_WATCH:
			w := new(pb.WatchReq)
			err = proto.Unmarshal(c.req.Data, w)
			if err != nil {
				c.writeErr(err)
				continue
			}
			c.serveWatch(w)
		}
		c.res.Reset()
		c.res.Seq = c.req.Seq
		c.writeResp(&c.res)

	}

}

func (c *client) writeResp(p *pb.Response) {
	c.mx.Lock()
	defer c.mx.Unlock()
	w, err := c.conn.NextWriter(websocket.BinaryMessage)
	if err == nil {
		bufP := bufPool.Get().([]byte)
		bufP = bufP[:0]
		bufP = protowire.AppendVarint(bufP, uint64(c.marshal.Size(p)))
		bufP, err = c.marshal.MarshalAppend(bufP, p)
		_, err = w.Write(bufP)
	}
	if err != nil {
		log.Println(err)
	}
}

func (c *client) writeErr(err error) {
	c.res.Reset()
	c.res.Msg = err.Error()
	c.res.Code = 500
	c.writeResp(&c.res)
}

func (c *client) Close() (err error) {
	err = c.conn.Close()
	return
}

func (c *client) serveWatch(req *pb.WatchReq) {
	prev := c.cancels[0]
	if prev != nil {
		prev()
		delete(c.cancels, 0)
	}
	w := c.s.dirs[req.Service]
	var fw *FileWatch
	if w != nil {
		if req.Id >= 0 && int(req.Id) < len(w.files) {
			fw = w.files[req.Id]
			id := int(req.Id)
			w.e <- &watchCtx{client: c, id: id, unWatch: req.UnWatch}
			if !req.UnWatch {
				c.cancels[0] = func() {
					w.e <- &watchCtx{client: c, id: id, unWatch: true}
				}
			}
		}
	}
	if fw == nil {
		c.writeErr(unknownService)
		return
	}
}

type LogServer struct {
	Em      *slog.Emitters
	clients map[*client]bool
	evt     chan Event
	dirs    map[string]*DirWatch
	ctx     context.Context
	cancel  context.CancelFunc
}

func NewLogServer(c *config.ServerConfig, em *slog.Emitters) (s *LogServer, err error) {
	s = &LogServer{
		Em:      em,
		clients: make(map[*client]bool),
		evt:     make(chan Event),
		dirs:    make(map[string]*DirWatch),
	}
	err = s.init(c)
	return
}

func (l *LogServer) ServeWs(w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Println(err)
		return
	}
	c := &client{s: l, conn: conn, cancels: make(map[int32]Cleaner)}
	go c.readMsg()
	c.writeMsg()
}

func (l *LogServer) ServeWatch(w http.ResponseWriter, r *http.Request) {
	idx := strings.LastIndex(r.URL.Path, "/") + 1
	if idx > 0 && len(r.URL.Path) > idx {
		path := r.URL.Path[idx:]
		wt := l.dirs[path]
		if wt != nil {
			rest.RenderJSON(w, wt.Dir.Items)
			return
		}
	}
	http.Error(w, "err req", http.StatusBadRequest)
}

func (l *LogServer) init(c *config.ServerConfig) (err error) {
main:
	for i, v := range c.Dirs {
		var w *DirWatch
		w, err = NewDirWatch(v, c)
		if err == nil {
			w.Id = i
			l.dirs[v.Name] = w
			for j, f := range v.Items {
				tools.Log.Println("init logger", v.Dir, f.Pattern)
				var (
					m  slog.MsgMaker
					fw *FileWatch
				)
				f.Id = j
				if f.Type == "java" {
					m = fex.NewMakeLog(&c.Java, v.Name)
				} else {
					m = &slog.LineMsgMaker{Name: v.Name}
				}

				collect := slog.NewCollect(&slog.Option{
					MaxBuf: c.MaxBuf,
				}, m, nil)

				fw, err = NewFileWatch(f, collect)

				if l.Em != nil {
					collect.Emitter = slog.NewEmitters(collect.Emitter, l.Em)
				}

				if f.Type == "java" && len(c.Java.Mybatis) > 0 {
					collect.Emitter = java.NewMybatis(collect.Emitter, c.Java.Mybatis)
				}

				if err == nil {
					fw.Id = j
					fw.w = w
					w.files[j] = fw
				} else {
					log.Println(err)
					break main
				}
			}
		} else {
			break
		}
	}
	return
}

func (l *LogServer) Run() {
	for _, v := range l.dirs {
		go v.Run()
	}

	l.ctx, l.cancel = context.WithCancel(context.Background())
main:
	for {
		select {
		case e, ok := <-l.evt:
			if !ok {
				return
			}
			switch e.GetType() {
			case evtReg:
				c := e.(*client)
				if c.s != nil {
					l.clients[e.(*client)] = true
				} else {
					delete(l.clients, c)
					err := c.Close()
					if err != nil {
						log.Println("client close err", err)
					}
				}
			}
		case <-l.ctx.Done():
			break main
		}

	}
}

func (l *LogServer) Close() (err error) {
	for _, v := range l.dirs {
		err = v.Close()
		if err != nil {
			log.Println(err)
		}
	}
	for c := range l.clients {
		err = c.Close()
	}

	if l.cancel != nil {
		l.cancel()
	}

	return
}

func (l *LogServer) ServeDir(r chi.Router) {
	r.Get("/{name}", l.listDir)
}

func (l *LogServer) listDir(w http.ResponseWriter, r *http.Request) {
	name := chi.URLParam(r, "name")
	f := l.dirs[name]
	if f == nil {
		w.WriteHeader(404)
		return
	}
	rs := make([]*WatchFile, 0)
	files, err := os.ReadDir(f.Dir.Dir)

	for _, info := range files {
		if !info.IsDir() && (strings.HasSuffix(info.Name(), ".log") || strings.HasSuffix(info.Name(), ".log.gz")) {
			logf, err := info.Info()
			if err != nil {
				log.Println(err)
				continue
			}
			rs = append(rs, &WatchFile{
				Name: info.Name(),
				Size: logf.Size(),
				Time: db.Time(logf.ModTime()),
			})
		}
	}

	if len(rs) > 0 {
		sort.Slice(rs, func(i, j int) bool {
			return time.Time(rs[i].Time).After(time.Time(rs[j].Time))
		})
	}
	if err != nil && err != fs.SkipDir {
		rest.RenderErr(w, err)
	}
	rest.RenderJSON(w, rs)
}
